package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.entity.ApArticle;
import com.heima.article.dto.UpdateArticleMessage;
import com.heima.article.service.IApArticleService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class BehaviorListener {
    @Autowired
    private IApArticleService apArticleService;

    /**
     * 点赞行为监听
     * @param consumerRecord
     */
    @KafkaListener(topics = "${topic.like_behavior}") //点赞行为监听
    public void handlerLikesMsg(ConsumerRecord<String,String> consumerRecord) {

        String value = consumerRecord.value();
        UpdateArticleMessage updateMessage = JSON.parseObject(value, UpdateArticleMessage.class);
        System.out.println("监听到消息,即将执行文章: "+updateMessage.getArticleId()+" 点赞数量"+updateMessage.getAdd()+"修改操作!!!");
        ApArticle article = apArticleService.getById(updateMessage.getArticleId());
        if (updateMessage.getType() == 1){//操作类型 0 阅读 1 点赞 2 评论 3 收藏
            if (updateMessage.getAdd()< 0 && article.getLikes()<=0){
                return;
            }
            article.setLikes(article.getLikes()+updateMessage.getAdd());
            apArticleService.updateById(article);
        }
    }

    /**
     * 阅读行为监听
     * @param consumerRecord
     */
    @KafkaListener(topics = "${topic.read_behavior}")
    public void handlerReadMsg(ConsumerRecord<String,String> consumerRecord) {
        String value = consumerRecord.value();

    }

}
